<?php

namespace app\command;

use think\console\Command;
use think\console\Input;
use think\console\Output;
use Swoole\Server;
use think\facade\Cache;

class MqttServer extends Command
{
    protected $subscriptions = []; // 用于存储客户端订阅信息

    protected function configure()
    {
        $this->setName('mqtt:server')->setDescription('Start MQTT Server');
    }

    protected function execute(Input $input, Output $output)
    {
        $server = new Server('0.0.0.0', 10036);

        $server->set([
            'worker_num' => 4,
            'log_file' => '/tmp/swoole.log',
        ]);

        $server->on('connect', function ($server, $fd) {
            echo "Client connected: $fd\n";
            $this->send($server,$fd);
        });

        $server->on('receive', function ($server, $fd, $reactorId, $data) {
            echo "Received data from $fd: ".bin2hex($data)."\n";
            $packetType = ord($data[0]) >> 4;
            echo "当前type:".$packetType."\n";
            // 处理订阅（SUBSCRIBE）报文
            // 这里需要解析 MQTT 数据，判断是订阅请求还是普通消息
            if ($packetType === 8) { // SUBSCRIBE 报文类型为 0x08
                $packetIdentifier = unpack('n', substr($data, 2, 2))[1];

                // 解析主题和 QoS
                $payload = substr($data, 4);
                $topics = [];
                while (strlen($payload) > 0) {
                    // 解析主题长度
                    $topicLength = unpack('n', substr($payload, 0, 2))[1];
                    $topicName = substr($payload, 2, $topicLength);
                    $qos = ord($payload[2 + $topicLength]); // QoS 是主题后面的一个字节
                    $this->subscriptions[$topicName][] = $fd;
                    $topics[] = ['topic' => $topicName, 'qos' => $qos];
                    echo "Client #{$fd} subscribed to topics: " . ($topicName);
                    // 截断已解析的部分
                    $payload = substr($payload, 2 + $topicLength + 1);
                }
                Cache::set("all_mqtt_topic",json_encode($this->subscriptions));
                // 构造 SUBACK 报文
                $subackPayload = '';
                foreach ($topics as $topic) {
                    // 假设所有订阅都成功，返回与客户端请求的 QoS 相同
                    $subackPayload .= chr($topic['qos']);
                }
                $suback = chr(0x90) . chr(2 + strlen($subackPayload)) . pack('n', $packetIdentifier) . $subackPayload;

                // 发送 SUBACK 报文
                $server->send($fd, $suback);
            }elseif($packetType===3){
                if (strlen($data) < 3) {
                    echo "Received PUBLISH message is too short.\n";
                    return; // 数据长度不足，返回
                }
                $remainingLength = ord($data[1]); // 剩余长度
                if (strlen($data) < 4 + $remainingLength) {
                    echo "Received PUBLISH message is incomplete.\n";
                    return; // 数据不完整，返回
                }
                $topicLength = ord($data[2]) << 8 | ord($data[3]); // 获取主题长度
                $topic = substr($data, 4, $topicLength); // 主题
                $messageStart = 4 + $topicLength; // 消息开始位置

                // 如果 QoS > 0，解析 Packet Identifier
                $qos = ($data[0] & 0x06) >> 1; // 获取 QoS
                $packetId = null;
                if ($qos > 0) {
                    $packetId = ord($data[$messageStart]) << 8 | ord($data[$messageStart + 1]); // Packet Identifier
                    $messageStart += 2; // 移动到消息内容
                }

                // 获取消息内容
                $message = substr($data, $messageStart, $remainingLength - $topicLength - ($qos > 0 ? 2 : 0) - 2);
                echo "  Message: {$message}\n";
                $message = json_decode($message, true);
                if (isset($message['topic']) && isset($message['message'])) {
                    // 遍历所有连接的客户端并推送消息
                    $this->publish($message['topic'],$message['message']);
                }
                //$this->send($server,$fd);
            }
        });

        $server->on('close', function ($server, $fd) {
            echo "Client closed: $fd\n";
            // 客户端断开连接时，移除其订阅信息
            foreach ($this->subscriptions as $topic => $clients) {
                $this->subscriptions[$topic] = array_filter($clients, function ($clientFd) use ($fd) {
                    return $clientFd != $fd;
                });
            }
        });

        $server->start();
    }

    // 推送消息到指定主题
    public function publish($topic, $message)
    {
        global $server;
        $this->subscriptions=json_decode(Cache::get('all_mqtt_topic'),true);
        echo "当前topic:".$topic;
        if (isset($this->subscriptions[$topic])) {
            foreach ($this->subscriptions[$topic] as $fd) {
                $server->send($fd, $message);
            }
        }
    }

    private function send($server,$fd) {
        echo '返回';
        $connack = chr(0x20) . chr(0x02) . chr(0x00) . chr(0x00);
        $server->send($fd, $connack);
    }
}